package com.kafka.demo.config;

import com.kafka.demo.config.yml.CommonKafkaConsumerConfig;
import com.kafka.demo.config.yml.KafkaConsumerProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ErrorHandler;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import java.util.HashMap;
import java.util.Map;

@Configuration
@Slf4j
public class KafkaConsumerConfig {

    @Autowired
    private CommonKafkaConsumerConfig commonConsumerConfig;

    @Autowired
    private KafkaConsumerProperties properties;

    private Map<String, Object> buildCommonConsumerProps() {
        return commonConsumerConfig.toConsumerProps();
    }

    // ========== 创建 ConsumerFactory ==========

    /**
     * 集群消费者1
     * @return
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> clusterListenerFactory1( ) {
        String consumerName = "cluster-consumer1";
        KafkaConsumerProperties.ConsumerConfig config = properties.getServices().get( consumerName);
        return factory(consumerName, config);
    }

    /**
     * 集群消费者2
     * @return
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> clusterListenerFactory2( ) {
        String consumerName = "cluster-consumer2";
        KafkaConsumerProperties.ConsumerConfig config = properties.getServices().get( consumerName);
        return factory(consumerName, config);
    }

    /**
     * 广播消费者1
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> broadcastListenerFactory1( ) {
        String consumerName = "broadcast-consumer1";
        KafkaConsumerProperties.ConsumerConfig config = properties.getServices().get( consumerName);
        return factory(consumerName, config);
    }

    /**
     * 广播消费者2
     */
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> broadcastListenerFactory2( ) {
        String consumerName = "broadcast-consumer2";
        KafkaConsumerProperties.ConsumerConfig config = properties.getServices().get( consumerName);
        return factory(consumerName, config);
    }

    public ConsumerFactory<String, String> consumerFactory(String consumerName) {
        Map<String, Object> props = buildCommonConsumerProps();
        props.put(ConsumerConfig.GROUP_ID_CONFIG,
                properties.getServices().get(consumerName).getGroupId());
        return new DefaultKafkaConsumerFactory<>(props);
    }
    private ConcurrentKafkaListenerContainerFactory<String, String> factory(String consumerName,KafkaConsumerProperties.ConsumerConfig config) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory(consumerName));
        factory.setConcurrency(config.getConcurrency());
        factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL_IMMEDIATE);
/*        factory.getContainerProperties().setErrorHandler(new ErrorHandler() {
            @Override
            public void handle(Exception e, ConsumerRecord<?, ?> record) {
                log.error("消费失败 topic={}, partition={}, 重试次数={}, value={}%n",
                        record.topic(), record.partition(), record.offset(), record.value());
                log.error(e.getMessage(),e);
            }
        });*/
        factory.setRetryTemplate(retryTemplate());
        return factory;
    }




    // ========== 重试组件 ==========

    public RetryTemplate retryTemplate() {
        RetryTemplate retryTemplate = new RetryTemplate();

        // 1. 重试策略：最多重试几次
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);

        // 2. 退避策略：重试之间的等待时间
        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(5000);  // 初始间隔
        backOffPolicy.setMultiplier(1.5);         // 乘数（倍数）
        backOffPolicy.setMaxInterval(20000);      // 最大间隔

        retryTemplate.setRetryPolicy(retryPolicy);
        retryTemplate.setBackOffPolicy(backOffPolicy);
        return retryTemplate;
    }


}